Skip to content

feat: Store only the last row in the previous cursor for round robin tie-breaking purposes#23619

Open
ariel-miculas wants to merge 5 commits into
apache:mainfrom
ariel-miculas:fix-23606
Open

feat: Store only the last row in the previous cursor for round robin tie-breaking purposes#23619
ariel-miculas wants to merge 5 commits into
apache:mainfrom
ariel-miculas:fix-23606

Conversation

@ariel-miculas

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Rationale for this change

See the linked issue.
Note that this PR also contains the changes in #23606, so this will have to be rebased

What changes are included in this PR?

Store only the last row in prev_cursors instead of keeping the entire cursor

Are these changes tested?

Added a test to show the improvement

Are there any user-facing changes?

No

/// any buffer the cursor's [`CursorValues`] holds. Used to remember a
/// partition's last row across a batch boundary without keeping the
/// whole exhausted batch's memory alive (see [`Self::is_eq_to_prev_one`]).
pub fn last_value(&self) -> T::SingleRowValue {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I feel like this should return an Option, in case values() is empty

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

values() shouldn't be empty, there are some asserts in the code:

        assert!(rows.num_rows() > 0);
        assert!(array.len() > 0, "Empty array passed to FieldCursor");

I don't imagine a Cursor with empty values being useful.

impl RowValues {
    /// Create a new [`RowValues`] from `rows` and a `reservation`
    /// that tracks its memory. There must be at least one row
    ///
    /// Panics if the reservation is not for exactly `rows.size()`
    /// bytes or if `rows` is empty.
    pub fn new(rows: Arc<Rows>, reservation: MemoryReservation) -> Self {
        assert_eq!(
            rows.size(),
            reservation.size(),
            "memory reservation mismatch"
        );
        assert!(rows.num_rows() > 0);
        Self {
            rows,
            _reservation: reservation,
        }
    }
}
    pub fn new<A: CursorArray<Values = T>>(
        options: SortOptions,
        array: &A,
        reservation: MemoryReservation,
    ) -> Self {
        assert!(array.len() > 0, "Empty array passed to FieldCursor");
        let null_threshold = match options.nulls_first {
            true => array.null_count(),
            false => array.len() - array.null_count(),
        };

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The previous code also assumed values is never empty:

    fn is_eq_to_prev_row_in_prev_batch(&self, other: &Self) -> bool {
        assert_eq!(self.offset, 0);
        T::eq(
            &self.values,
            self.offset,
            &other.values,
            other.values.len() - 1,
        )
    }

@ariel-miculas

Copy link
Copy Markdown
Contributor Author

@Dandandan wondering whether we can change ReusableRows to only hold one slot after this change, since the previous cursor will no longer need to be kept alive for the round robin tie-breaking feature.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

core Core DataFusion crate physical-plan Changes to the physical-plan crate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants